import java.sql.{DriverManager, PreparedStatement}

import org.apache.spark.{SparkConf, SparkContext}

object Data2MysqlForeach {
  def main(args:Array[String]) : Unit = {
    val sparkConf = new SparkConf().setAppName("Data2MysqlForeach").setMaster("local[2]")
    val sc = new SparkContext(sparkConf)

    val data = sc.textFile("data/access.log")


    val personRDD = data.map(x => x.split(",")).map(x => (x(0),x(1),x(2).toInt))

    personRDD.foreach(line => {
      val connection = DriverManager.getConnection("jdbc:mysql://node03:3306/spark","root","123456")
      val sql="insert into person(id,name,age) values(?,?,?)"
      try {
        val ps: PreparedStatement = connection.prepareStatement(sql)

        //4、获取数据,给？号 赋值
        ps.setString(1, line._1)
        ps.setString(2, line._2)
        ps.setInt(3, line._3)
        //执行
        ps.execute()
      } catch {
        case e:Exception => e.printStackTrace()
      } finally {
        if (connection != null) {
          connection.close()
        }
      }
    })
  }
}
